package com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.kafka.streams;

import com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.streams.JsonSerde;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.*;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.WindowStore;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;

import java.time.Duration;

@Configuration
@EnableKafka
public class DashboardKafkaStreams {

    @Bean
    public KStream<Windowed<String>, AgentStatistic> agentStatisticKStream(StreamsBuilder myKStreamBuilder) {
        KTable<String, AgentProfile> agentProfiles = myKStreamBuilder
                .table("agent-profiles", Consumed.with(Serdes.String(), new JsonSerde(AgentProfile.class)), Materialized.<String, AgentProfile, WindowStore<Bytes, byte[]>>as("agent-profile-state").withValueSerde(new JsonSerde(AgentProfile.class)));
        KStream<String, AgentSummary> agentSummaries = agentProfiles.toStream().map((agentId, agentProfile) -> new KeyValue(agentId, new AgentSummary(agentProfile)));
        KStream<String, AgentCase> agentCases = myKStreamBuilder.stream("agent-cases", Consumed.with(Serdes.String(), new JsonSerde(AgentCase.class)));
        KStream<String, AgentPresence> agentPresences = myKStreamBuilder.stream("agent-presences", Consumed.with(Serdes.String(), new JsonSerde(AgentPresence.class)));
        agentSummaries = agentSummaries.leftJoin(agentCases,
                (agentId, agentSummary, agentCase) -> {
                    if (null != agentCase) {
                        agentSummary.updateCaseInfo(agentCase);
                    }
                    return agentSummary;
                },
                JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)).before(Duration.ofSeconds(0)),
                StreamJoined.with(Serdes.String(), new JsonSerde<>(AgentSummary.class), new JsonSerde<>(AgentCase.class)));
        agentSummaries.print(Printed.toSysOut());
        agentSummaries = agentSummaries.leftJoin(agentPresences,
                        (agentId, agentSummary, agentPresence) -> {
                            if (null != agentPresence) {
                                agentSummary.updatePresence(agentPresence);
                            }
                            return agentSummary;
                        },
                        JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)).before(Duration.ofSeconds(0)),
                        StreamJoined.with(Serdes.String(), new JsonSerde<>(AgentSummary.class), new JsonSerde<>(AgentPresence.class)))
                .map(((key, value) -> new KeyValue<>(value.getAgentId(), value)));
        agentSummaries.print(Printed.toSysOut());
        KStream<Windowed<String>, AgentStatistic> agentStatistics = agentSummaries
                .groupByKey(Grouped.with(Serdes.String(), new JsonSerde<>(AgentSummary.class)))
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(60)).advanceBy(Duration.ofSeconds(3600)))
                .aggregate(AgentStatistic::new,
                        (agentId, agentSummary, agentStatistic) -> {
                            agentStatistic.setAgentId(agentSummary.getAgentId());
                            agentStatistic.setAgentName(agentSummary.getAgentName());
                            agentStatistic.getCaseHandling().addAll(agentSummary.getCaseHandling());
                            agentStatistic.getCaseHandled().addAll(agentSummary.getCaseHandled());
                            agentStatistic.setStatus(agentSummary.getStatus());
                            agentStatistic.setStatusDuration(agentSummary.getStatusDuration());
                            agentStatistic.setConfiguredCapacity(agentSummary.getConfiguredCapacity());
                            return agentStatistic;
                        },
                        Materialized.<String, AgentStatistic, WindowStore<Bytes, byte[]>>as("agent-aggregates").withKeySerde(Serdes.String()).withValueSerde(new JsonSerde(AgentStatistic.class)))
                .toStream();
        agentStatistics.print(Printed.toSysOut());
        agentStatistics.to("agent-statistics");
        return agentStatistics;
    }
}
